// Copyright (c) 2022 Presto Labs Pte. Ltd.
// Author: qiang

#include "coin2/exchange/base/order/sync_trade_client.h"

#include <memory>
#include <string>
#include <utility>

#include "coin2/exchange/base/order/external.h"
#include "coin2/exchange/base/order/sync_trade_client.h"
#include "coin2/exchange/base/symbology/util.h"
#include "coin2/exchange/factory/order/system_components.h"
#include "coin2/exchange/gdax/api_util/transfers_account_store.h"
namespace coin2::exchange::base::order {
using coin2::exchange::factory::order::OrderSubsystemComponents;
using coin2::exchange::order::OrderSystemConfig;
using coin2::exchange::order::PrivateSubscriberConfig;
using RestQueryProto = coin::proto::RestQueryProto;
using OrderConnectionConfig = coin2::exchange::order::OrderConnectionConfig;
using coin2::exchange::gdax::api_order::GdaxTransfersAccountStore;
using AsioHttpContextList = std::list<std::shared_ptr<AsioHttpContext>>;

static std::map<std::string, int> once_query_max_map = {{"Spot.Gateio.v4", 30},
                                                        {"Spot.Kucoin.v1", 1},
                                                        {"Spot.Mexc.v3", 10},
                                                        {"Spot.Binance.v1", 30},
                                                        {"Spot.Okex.v5", 30},
                                                        {"Spot.Gdax.v1", 90},
                                                        {"Spot.Huobi.v1", 10},
                                                        {"Spot.Phemex.v1", 2000},
                                                        {"Futures.Deribit.v2", 2000}};

SyncTradeClient::SyncTradeClient(const coin2::exchange::order::OrderSystemConfig& osc_config)
    : osc_config_(osc_config) {
  if (once_query_max_map.count(osc_config_.mea()) == 0) {
    throw std::runtime_error("Unsupported exchange: " + osc_config_.mea());
  }

  mea_ = MarketExchangeApi::FromString(osc_config_.mea());

  OrderSystemConfig sys_config;
  PrivateSubscriberConfig ps_config;
  boost::asio::io_context ioc;
  auto driver = std::make_unique<NativeHttpAsioDriver>(&ioc, mea_, osc_config_.connection_config());
  *sys_config.mutable_connection_config() = osc_config_.connection_config();
  OrderSubsystemComponents component;
  component.Create(mea_, sys_config, ps_config, nullptr, driver.get());
  client_ = std::move(component.rest_client());
  client_->SetMessageBus(&bus_);
  parser_ = std::move(component.parser());
  parser_->SetTransferFirstRecord();
  parser_->SetMessageBus(&bus_);
}

std::vector<std::pair<int64_t, int64_t>>
GetTimeSection(const int64_t start_timestamp, const int64_t end_timestamp, std::string mea) {
  size_t limit_day = once_query_max_map[mea];
  const int64_t day_ns = 24 * 60 * 60 * 1e9;
  int64_t query_section_time = day_ns * limit_day;
  std::vector<std::pair<int64_t, int64_t>> sections;
  int64_t cur = end_timestamp - query_section_time;
  int64_t end = end_timestamp;
  int64_t start = start_timestamp;
  while (start < cur) {
    auto section = std::make_pair(cur, end);
    sections.emplace_back(section);
    end = cur;
    cur = end - query_section_time;
  }
  sections.emplace_back(std::make_pair(start, end));

  return sections;
}

AsioHttpContextList SyncTradeClient::GetRequestList(
    const int64_t start_timestamp,
    const int64_t end_timestamp,
    std::string mea) {
  AsioHttpContextList request_list;

  RestQueryProto query;
  query.set_start_time(start_timestamp);
  query.set_end_time(end_timestamp);
  query.set_type(RestQueryProto::QUERY_WITHDRAW);
  request_list.merge(client_->SubmitRestQueryList(query));
  query.set_type(RestQueryProto::QUERY_DEPOSIT);
  request_list.merge(client_->SubmitRestQueryList(query));

  if (mea == "Spot.Gdax.v1") {
    const auto& id_currency_map = GdaxTransfersAccountStore::GetInstance().GetCurrencyIdMap();
    for (const auto& [id, _] : id_currency_map) {
      query.set_gdax_currency_id(id);
      request_list.merge(client_->SubmitRestQueryList(query));
    }
  }

  query.set_type(RestQueryProto::QUERY_TRANSFERS);
  request_list.merge(client_->SubmitRestQueryList(query));
  return request_list;
}

void SyncTradeClient::QueryTransferHistories(
    const int64_t start_timestamp,
    const int64_t end_timestamp,
    std::vector<coin::proto::CurrencyTransferElement>* transfer_out) {
  Init();
  std::vector<coin::proto::CurrencyTransferElement> result;

  const auto& sections = GetTimeSection(start_timestamp, end_timestamp, osc_config_.mea());
  for (int i = 0; i < sections.size(); i++) {
    auto section = sections[i];
    AsioHttpContextList request_list =
        GetRequestList(section.first, section.second, osc_config_.mea());

    for (auto& request : request_list) {
      TransferExecutor(request, &result);
      if (osc_config_.mea() == "Spot.Okex.v5") {
        usleep(200000);
      }
    }

    if (osc_config_.mea() == "Futures.Deribit.v2") {
      break;
    }
  }

  transfer_out->clear();
  for (int i = 0; i < result.size(); i++) {
    if (result[i].create_timestamp() >= start_timestamp &&
        result[i].create_timestamp() <= end_timestamp) {
      transfer_out->emplace_back(std::move(result[i]));
    }
  }
}

void SyncTradeClient::TransferExecutor(
    std::shared_ptr<AsioHttpContext> request,
    std::vector<coin::proto::CurrencyTransferElement>* transfer_out) {
  boost::asio::io_context ioc;
  auto driver = std::make_unique<NativeHttpAsioDriver>(&ioc, mea_, osc_config_.connection_config());
  request->handler = [&](std::shared_ptr<AsioHttpContext> context) {
    BaseParser::RawMessage msg;
    msg.http_context = context;
    msg.timestamp = context->response_timestamp;
    msg.type = coin::proto::REST_MESSAGE;
    auto result = this->parser_->ParseRawMessage(msg);
    if (result.has_error()) {
      if (transfer_out) {
        transfer_out->clear();
      }
      ioc.stop();
      throw std::runtime_error("Parsing Error: " + msg.SerializeToString());
    }
    if (transfer_out) {
      auto currency_transfer = result.mutable_currency_transfer_bundle()->each_transfer();
      for (const auto& transfer_item : currency_transfer) {
        transfer_out->emplace_back(transfer_item);
      }
    }
    ioc.stop();
  };
  driver->PostAsyncContext(request);
  ioc.run();
}

void SyncTradeClient::Init() {
  if (GetCurrentTimestamp() - last_init_timestamp_ > 300 * 1'000'000'000LL &&
      osc_config_.mea() == "Futures.Deribit.v2") {
    is_initialized = false;
  }

  if (is_initialized) {
    return;
  }

  if (osc_config_.mea() == "Spot.Gdax.v1") {
    if (GdaxTransfersAccountStore::GetInstance().is_initialized()) {
      is_initialized = true;
      return;
    }
  }

  if (osc_config_.mea() == "Spot.Gdax.v1" || osc_config_.mea() == "Spot.Huobi.v1" ||
      osc_config_.mea() == "Futures.Deribit.v2") {
    RestQueryProto query;
    query.set_type(RestQueryProto::INIT_QUERY_INFO);
    auto request_list = client_->SubmitRestQueryList(query);
    for (auto& item : request_list) {
      TransferExecutor(item, nullptr);
    }
  }

  last_init_timestamp_ = GetCurrentTimestamp();
  is_initialized = true;
  return;
}

}  // namespace coin2::exchange::base::order
